package com.ewei.web.crawler.dao;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.bson.Document;

import com.ewei.web.crawler.domain.URLs;
import com.ewei.web.crawler.utils.BaseDao;
import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.MongoBulkWriteException;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCursor;
/**
 * 资源链接队列
 * 考虑到比起一条条读取,可能大数据量读取性能更佳
 * 所以内部做了个只读的缓存,尽可能一次读取大量数据到缓存中
 * 2016 8/13
 * 把mongodb的cursor当作缓存的指针,另开一个线程来同步修改过的缓存数据到数据库
 * cursor(游标)可以做到分批读取,不担心内存爆炸问题
 * 2016 8/14
 * @author David
 *
 */
public class URLQueue  extends BaseDao<URLs> implements Queue<URLs>{
	private static Queue<URLs> queue = null;
	//设置只读的缓冲区
	private Cache<Document> cache = null;
	private URLQueue(){
		cache = new InnerCache();
	}
	public static Queue<URLs> getInstance(){
		if(null == queue){
			synchronized(URLQueue.class){
				if(null == queue){
					queue = new URLQueue();
				}
			}
		}
		return queue;
	}
	/**
	 * 初始化队列(数据库存储的)
	 */
	public void init() {
		//先销毁缓冲区
		cache.destory();
		//担心内存溢出
		System.gc();
		cache = new InnerCache();
		collection.updateMany(
				new BasicDBObject(), 
				new BasicDBObject("$set",new BasicDBObject("status",0))
		);
	}
	/**
	 * 当前队列大小(数据库存储的)
	 */
	public long size() {
		return size0();
	}
	private long size0(){
		BasicDBObject bson = new BasicDBObject("status",0);
		return collection.count(bson);
	}
	/**
	 * 直接推入数据库队列(暂不加缓存,用唯一索引去重)
	 */
	public void push(URLs t) {
		try{
			t.setQueueTime(System.currentTimeMillis());
			super.insert(t);
		}catch(Exception e){//不处理写异常
		}
	}
	/**
	 * 出列(从缓存中)
	 */
	public URLs poll() {
		return documentToURLs(cache.read());
	}
	
	private URLs documentToURLs(Document document) {
		if(null != document){
			URLs topic = new URLs(document.getString("url"),document.getString("description"));
			topic.setId(document.getObjectId("_id").toHexString());
			topic.setStatus(document.getInteger("status"));
			topic.setQueueTime(document.getLong("queueTime"));
			return topic;
		}
		return null;
	}
	
	public boolean isEmpty() {
		return cache.isEmpty();
	}
	@Override
	public void repush(String id) {
		Document document0 = new Document("_id",id);
		Document document1 = new Document("$set",new Document("status",0));
		collection.findOneAndUpdate(document0, document1);
	}
	/**
	 * 内部缓存
	 * @author David
	 *
	 * @param <URLs>
	 */
	private class InnerCache implements Cache<Document>{
		private ReentrantLock cache = new ReentrantLock();
		private ReentrantLock sync = new ReentrantLock();
		private Condition notEmpty = sync.newCondition();
		private Condition finish = sync.newCondition();
		private MongoCursor<Document> cursor = null;
		private List<Document> datas = new ArrayList<Document>();
		private SyncThread<Document> syncThread = null;
		public InnerCache(){
			syncThread = new SyncThread<>(this);
			new Thread(syncThread).start();//开启同步线程
		}
		/**
		 * 从缓冲区中读取数据,若无则从数据库中取
		 */
		@Override
		public  Document read() {
			Document doc = null;
			cache.lock();
			try{
				if(isEmpty()){
					return null;
				}
				doc = cursor.next();
				doc.put("queueTime", System.currentTimeMillis());
			}finally{
				cache.unlock();
			}
			sync.lock();
			try{
				datas.add(doc);
				notEmpty.signal();
			}finally{
				sync.unlock();
			}
			return doc;
		}
		@Override
		public void write(Document t) {
			throw new IllegalArgumentException("the innerCache doesn't have write method");
		}
		/**
		 * 同步缓存
		 */
		@Override
		public void sync() {
			sync.lock();
			try{
				if(datas.size() <= 0){
					notEmpty.await();
				}
				Iterator<Document> iterator = datas.iterator();
				while(iterator.hasNext()){
					Document doc = iterator.next();
					collection.findOneAndUpdate(
						new BasicDBObject("_id",doc.get("_id")), 
							new BasicDBObject("$set",
									new BasicDBObject("queueTime",doc.get("queueTime")).append("status", 1)));
				}
				datas.clear();
				finish.signalAll();
			}catch(InterruptedException e){
			}finally{
				sync.unlock();
			}
		}
		/**
		 * 销毁缓存
		 */
		@Override
		public void destory() {
			this.syncThread.close();
		}
		/**
		 * 检查缓存数据是否为空
		 * @return
		 */
		public boolean isEmpty(){
			if(cursor == null ||!cursor.hasNext()){//为空则检查数据库
				return check();
			}
			return false;
		}
		/**
		 * 检查数据库数据是否为空,同时更新cursor
		 */
		private boolean check(){
			sync.lock();
			try{
				if(datas.size()>0){
					finish.await();
				}
				cursor = getCursor();
			}catch(InterruptedException e){
			}	
			finally{
				sync.unlock();
			}
			return !cursor.hasNext();
		}
		/**
		 * 从数据库中查询数据,获得cursor
		 * @param size cursor的大小
		 */
		private MongoCursor<Document> getCursor(){
			if(null!=cursor){
				cursor.close();
			}
			BasicDBObject bson0 = new BasicDBObject();
			bson0.append("status", 0);
			bson0.append("queueTime",
						new BasicDBObject("$lte",System.currentTimeMillis()));
			FindIterable<Document> documents = collection.find(bson0);
			MongoCursor<Document> cursor= documents.iterator();
			return cursor;
		}
	}
}
